package org.leopard.rpc;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

/**
 * 服务中心
 * @author IBM CHEN ZHI KAN
 * @create 2018/8/13
 */
public class ServiceCenter {
  public static ServiceCenter instance;
  private ZooKeeper zk;
  final CountDownLatch connectedSignal=new CountDownLatch(1);
  //路由远程服务地址 协议url:远程服务地址
  public static Map<String,Set<String>> addres=new HashMap<String,Set<String>>();
  private ServiceCenter(String host){
    try {
      zk=connect(host);
    } catch (IOException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
  public static ServiceCenter getInstance(String host) {
    if(instance==null){
      instance=new ServiceCenter(host);
    }
      return instance;
  }
  /**
   * 判断节点是否存在
   * @param path
   * @return
   * @throws KeeperException
   * @throws InterruptedException
   */
  private Stat znodeExist(String path) throws KeeperException, InterruptedException {
    return zk.exists(path,true);
  }
  /**
   * 连接zk
   * @param host
   * @return
   * @throws IOException
   * @throws InterruptedException
   */
  private ZooKeeper connect(String host) throws IOException, InterruptedException {
    zk=new ZooKeeper(host, 5000, new Watcher() {
      public void process(WatchedEvent we) {
        if(we.getState()== KeeperState.SyncConnected){
          connectedSignal.countDown();
        }
      }
    });
    connectedSignal.await();
    return zk;
  }

  /**
   * 释放zk连接
   * @throws InterruptedException
   */
  private void destory() throws InterruptedException {
    zk.close();
  }

  /**
   * 创建节点
   * @param path
   * @param data
   * @Ids Ids.OPEN_ACL_UNSAFE 任何人都可以读写 CREATOR_ALL_ACL 只有创建者可以读写 READ_ACL_UNSAFE 任何人只能读
   * @CreateMode CreateMode.PERSISTENT 临时节点 PERSISTENT_SEQUENTIAL 临时有序节点 EPHEMERAL永久节点 EPHEMERAL_SEQUENTIAL永久有序节点
   * @throws KeeperException
   * @throws InterruptedException
   */
  public  void createNode(String path,byte[] data,List<ACL> ids,CreateMode createMode) throws KeeperException, InterruptedException {
    zk.create(path,data,ids,createMode);
  }

  /**
   * 移除节点
   * @param path
   * @throws KeeperException
   * @throws InterruptedException
   */
  public  void removeNode(String path) throws KeeperException,InterruptedException {
    zk.delete(path,zk.exists(path,true).getVersion());
  }

  public boolean createChildNode(String path,byte[] data,List<ACL> ids,CreateMode createMode){
    try {
      Stat stat=znodeExist(RPCContext.RPC_NODE_PATH);
      if(stat==null){
        createNode(RPCContext.RPC_NODE_PATH,"".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
      }
      createNode(path,data,Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
      return true;
    } catch (KeeperException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return false;
  }

  public void getChildNodes(String path){
    try {
      zk.getChildren(path, new Watcher() {
        public void process(WatchedEvent watchedEvent) {
          if(watchedEvent.getType().equals(EventType.NodeChildrenChanged)){

          }
        }
      });
    } catch (KeeperException e) {
      e.printStackTrace();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }


  public  void modifyData(String path, byte[] data) throws
      KeeperException,InterruptedException {
    zk.setData(path, data, zk.exists(path,true).getVersion());
  }

  public  byte[] getData(String path) throws Exception {
    if(znodeExist(path)!=null){
      return zk.getData(path,
          false, null);
    }else {
      throw new Exception(path+" node is not exist");
    }
  }
}
